package com.ruoyi.web.websocket;

import com.ruoyi.system.service.system.ISysUserService;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.timeout.IdleStateEvent;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.http.util.TextUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.MultiValueMap;
import org.yeauty.annotation.*;
import org.yeauty.pojo.Session;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * @创建者：zhouwei
 * @创建时间：2021/3/2
 * @描述： websocket 服务端
 */
@Component
@ServerEndpoint(path = "${ws.path}",port = "${ws.port}")
public class WebSocketServer {

    @Autowired
    private ISysUserService userService;

    Log log = LogFactory.getLog(WebSocketServer.class);
    //静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
    private static final int onlineCount = 0;
    //concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
    public static CopyOnWriteArraySet<MySession> webSocketSet = new CopyOnWriteArraySet<>();

    /**
     *  当有新的连接进入时，对该方法进行回调 注入参数的类型:Session、HttpHeaders
     * @param session
     * @param headers
     * @param req
     * @param reqMap
     * @param arg
     * @param pathMap
     */
    @BeforeHandshake
    public void handshake(Session session, HttpHeaders headers, @RequestParam String req,
                          @RequestParam MultiValueMap reqMap, @PathVariable String arg, @PathVariable Map pathMap){
        session.setSubprotocols("stomp");
        System.out.println("握手请求过来的数据" + req);
//        if (!"ok".equals(req)){
//            System.out.println("Authentication failed!");
//            session.close();
//        }
    }

    /**
     * 当有新的WebSocket连接完成时，对该方法进行回调 注入参数的类型:Session、HttpHeaders
     * @param session
     * @param headers
     * @param req
     * @param reqMap
     * @param arg
     * @param pathMap
     */
    @OnOpen
    public void onOpen(Session session, HttpHeaders headers, @RequestParam String req, @RequestParam MultiValueMap reqMap,
                       @PathVariable String arg, @PathVariable Map pathMap){
        try {
            this.sendMessage(session,"connect");
            String sid = pathMap.containsKey("sid") ? pathMap.get("sid").toString() : "";
            System.out.println("onOpen 有新连接：" + sid);
//            if(!TextUtils.isBlank(sid)){
//                sessionAdd(session,sid);
//                this.sendMessage(session,"connect");
//            }else{
//                this.sendMessage(session,"not have this user");
//            }
        } catch (IOException e) {
            log.error("websocket IO异常");
        }
    }

    /**
     *  新增连接
     * @param session
     * @param sid
     */
    private void sessionAdd(Session session,String sid){
        MySession mySession = new MySession(session,sid);
        Session session2 = getSessionBySid(sid);
        if(session2  == null){
            webSocketSet.add(mySession);
        }
        log.info("有新窗口开始监听:"+sid+", 当前在线人数为" + getOnLineUserCount());
    }

    /**
     *  移除连接
     * @param session
     */
    private void sessionRemove(Session session){
        for (MySession item : webSocketSet) {
            if(item.getSession().id().equals(session.id())){
                webSocketSet.remove(item);
                break;
            }
        }
        log.info("有一连接关闭！当前在线人数为" + getOnLineUserCount());
    }

    /**
     * 向单个用户推送
     */
    public void sendMessage(Session session, String message) throws IOException {
        session.sendText(message);
    }

    /**
     * 向所有用户推送
     */
    public void sendMessageToAll(String message) throws IOException {
        try {
            for (MySession item : webSocketSet) {
                Session session = item.getSession();
                session.sendText(message);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 发送自定义消息
     * */
    public void sendInfo(String message,String sid) throws IOException {
        log.info("推送消息到窗口" + sid + "，推送内容:" + message);
        try {
            if(!TextUtils.isBlank(sid)){//sid 不为空 只向当前用户推送消息
                Session session = getSessionBySid(sid);
                if(session!=null){
                    sendMessage(session,message);
                }
            }else{
                sendMessageToAll(message);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 当有WebSocket连接关闭时，对该方法进行回调 注入参数的类型:Session
     * @param session
     * @throws IOException
     */
    @OnClose
    public void onClose(Session session) throws IOException {
        System.out.println("one connection closed");
        sessionRemove(session);  //从set中删除
    }

    /**
     * 当有WebSocket抛出异常时，对该方法进行回调 注入参数的类型:Session、Throwable
     * @param session
     * @param throwable
     */
    @OnError
    public void onError(Session session, Throwable throwable) {
        log.error("发生错误");
        throwable.printStackTrace();
    }

    /**
     * 当接收到字符串消息时，对该方法进行回调 注入参数的类型:Session、String
     * @param session
     * @param message
     */
    @OnMessage
    public void onMessage(Session session, String message) {
        System.out.println("onMessage 111 " +message);
        try {
            if(!TextUtils.isBlank(message)){
                if(message.contains("connect|")){//链接信息
                    String sid = message.split("\\|")[1];
                    sessionAdd(session,sid);
                    this.sendMessage(session,"connect");
                }else{//正常的消息
                    String sid = getSidBySession(session);
                    if(!TextUtils.isBlank(sid)){
                        System.out.println("onMessage sid" + sid);
                        log.info("收到来自窗口"+sid+"的信息:"+message);
                        this.sendMessage(session,message);
                    }else{
                        this.sendMessage(session,"not has this user");
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 当接收到二进制消息时，对该方法进行回调 注入参数的类型:Session、byte[]
     * @param session
     * @param bytes
     */
    @OnBinary
    public void onBinary(Session session, byte[] bytes) {
        for (byte b : bytes) {
            System.out.println("onBinary:" +b);
        }
        session.sendBinary(bytes);
    }

    /**
     * 当接收到Netty的事件时，对该方法进行回调 注入参数的类型:Session、Object
     * @param session
     * @param evt
     */
    @OnEvent
    public void onEvent(Session session, Object evt) {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent idleStateEvent = (IdleStateEvent) evt;
            switch (idleStateEvent.state()) {
                case READER_IDLE:
                    System.out.println("read idle");
                    break;
                case WRITER_IDLE:
                    System.out.println("write idle");
                    break;
                case ALL_IDLE:
                    System.out.println("all idle");
                    break;
                default:
                    break;
            }
        }
    }

    /**
     * 遍历集合获取  sid
     * @param session
     * @return
     */
    private String getSidBySession(Session session){
        String sid = "";
        for (MySession item : webSocketSet) {
            System.out.println("sessionId = " + item.getSession().id());
            System.out.println("session id = " + session.id());
            if(item.getSession().id().equals(session.id())){
                sid = item.getSid();
                break;
            }
        }
        return sid;
    }

    /**
     * 遍历集合获取  session
     * @param sid
     * @return
     */
    private Session getSessionBySid(String sid){
        Session session = null;
        for (MySession item : webSocketSet) {
            if(item.getSid().equals(sid)){
                session = item.getSession();
                break;
            }
        }
        return session;
    }

    public static synchronized int getOnLineUserCount(){
        if(webSocketSet!=null){
            return webSocketSet.size();
        }
        return  0;
    }
}
